package com.atguigu.chapter07.D01_Window;

import com.atguigu.bean.WaterSensor;
import com.atguigu.util.AnqclnUtil;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Author: Pepsi
 * Date: 2023/8/4
 * Desc:
 */
public class Flink03_Window_Reduce_Process {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        env
                .socketTextStream("hadoop101", 9999)
                .map(line -> {
                    String[] data = line.split(",");

                    return new WaterSensor(
                            data[0],
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );
                })
                .keyBy(WaterSensor::getId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                .sum("vc")
                .reduce(new ReduceFunction<WaterSensor>() {
                            @Override
                            public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                                System.out.println("reduce 执行了...");
                                value1.setVc(value1.getVc() + value2.getVc());
                                return value1;
                            }
                        },
                        // 前面聚合函数的输出就是这个窗口处理函数的输入！！！
                        // 泛型：输入类型，输出类型，key的类型，窗口类型
                        new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                            @Override
                            public void process(String s,   // key
                                                Context context,   // 上下文
                                                Iterable<WaterSensor> elements,  // 这个集合中有且仅有一个元素：就是前面聚合的最终结果
                                                Collector<String> out) throws Exception {
                                System.out.println("process 执行了...");
                                WaterSensor result = elements.iterator().next();

                                String startTime = AnqclnUtil.toDateTime(context.window().getStart());
                                String endTime = AnqclnUtil.toDateTime(context.window().getEnd());

                                out.collect(startTime+" "+endTime+" "+result);
                            }
                        }
                )
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
